package com.networkflow_analysis

import java.sql.Timestamp
import java.text.SimpleDateFormat

import com.networkflow_analysis.bean.{ApacheLogEvent, PageViewCount}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/29 23:26
  *  热门页面浏览量统计
  *  无法处理迟到数据
  */
object HotPagesNetworkFlow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    //读取数据，转换成样例类提取时间戳和watermark
    val filePath: String = "NetworkFlowAnalysis/src/main/resources/apache.log"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val socketStream: DataStream[String] = env.socketTextStream("192.168.88.106",7777)


    val dataStream: DataStream[ApacheLogEvent] = socketStream.map(data => {
      val arr = data.split(" ")
      //对事件时间进行转换，得到时间戳
      val simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss")
      val ts = simpleDateFormat.parse(arr(3)).getTime
      ApacheLogEvent(arr(0), arr(1), ts, arr(5), arr(6))
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent](Time.seconds(1)) {
        override def extractTimestamp(t: ApacheLogEvent): Long = t.timestamp
      })

    //进行开窗聚合，以及排序输出
    val aggStream: DataStream[PageViewCount] = dataStream.filter(_.method == "GET")
      .filter(data => {
        val pattern = "^((?!\\.(css|js|ico|png)$).)*$".r
        (pattern findFirstIn data.url).nonEmpty
      })
      .keyBy(_.url)
      .timeWindow(Time.minutes(10), Time.seconds(5))
      .allowedLateness(Time.minutes(1))
      .sideOutputLateData(new OutputTag[ApacheLogEvent]("late"))
      .aggregate(new PageCountAgg, new PageViewCountWindowResult)

    val resultStream: DataStream[String] = aggStream.keyBy(_.windowEnd)
      .process(new TopNHotPages(3))

    dataStream.print("data")
    aggStream.print("agg")
    aggStream.getSideOutput(new OutputTag[ApacheLogEvent]("late")).print("late")
    resultStream.print("hot page")

    //输出

    env.execute("hot page")
  }

  class PageCountAgg extends AggregateFunction[ApacheLogEvent, Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(in: ApacheLogEvent, acc: Long): Long = acc + 1

    override def getResult(acc: Long): Long = acc

    override def merge(acc: Long, acc1: Long): Long = acc + acc1
  }

  class PageViewCountWindowResult extends WindowFunction[Long, PageViewCount, String, TimeWindow] {
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[PageViewCount]): Unit = {
      out.collect(PageViewCount(key, window.getEnd, input.iterator.next()))
    }
  }

  class TopNHotPages(n: Int) extends KeyedProcessFunction[Long, PageViewCount, String] {

    lazy val pageViewCountListState: ListState[PageViewCount] = getRuntimeContext
      .getListState(new ListStateDescriptor[PageViewCount]("pageViewCount-list", classOf[PageViewCount]))

    override def processElement(i: PageViewCount, context: KeyedProcessFunction[Long, PageViewCount, String]#Context, collector: Collector[String]): Unit = {
      pageViewCountListState.add(i)
      context.timerService().registerEventTimeTimer(i.windowEnd + 1)
    }

    //timestamp 定时器时间，ctx获取事件时间
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, PageViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
      val allPageViewCounts: ListBuffer[PageViewCount] = ListBuffer()
      val iter = pageViewCountListState.get().iterator()
      while (iter.hasNext) {
        allPageViewCounts += iter.next()
      }

      //提前清空状态
      pageViewCountListState.clear()
      //按照访问量排序并输出top n
      val sortedPageViewCounts = allPageViewCounts.sortWith(_.count > _.count)
        .take(n)

      //将排名信息格式化String，便于打印输出可视化
      val result: StringBuilder = new StringBuilder
      result.append("窗口结束时间：").append(new Timestamp(timestamp - 1)).append("\n")

      //遍历结果列表中的每个ItemViewCount，输出到一行
      for (i <- sortedPageViewCounts.indices) {
        val currentPageViewCount = sortedPageViewCounts(i)
        result
          .append("NO: ").append(i + 1).append("\t")
          .append("商品ID = ").append(currentPageViewCount.url).append("\t")
          .append("热门度 = ").append(currentPageViewCount.count).append("\n")
      }
      result.append("===============================================\n")

      Thread.sleep(1000)
      out.collect(result.toString())
    }
  }

}
